package com.cristina.attention.manager;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cristina.attention.codec.Invocation;
import com.cristina.attention.feign.GroupMemberFeignService;
import com.cristina.attention.message.chat.ChatRedirect;
import com.cristina.attention.utils.R;
import com.cristina.attention.utils.RedisUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 获取群成员可以考虑将群成员放入 reids,
 *
 * @Author superman
 * @Date 2021/9/21 20:57
 **/
@Component
@Slf4j
public class NettyChannelManager {

    public static final AttributeKey<Long> CHANNEL_ATTR_KEY_USER = AttributeKey.newInstance("user");

    @Resource
    private RedisUtils redisUtils;

    @Resource
    private GroupMemberFeignService groupMemberFeignService;

    /**
     * ChannelId: Channel
     */
    private ConcurrentMap<ChannelId, Channel> channels = new ConcurrentHashMap<>();

    /**
     * 暂时代替 redis 做本地缓存, TODO 使用 redis 缓存 Channel 连接 , 会出现获取key的时候id会跟着变
     * channel是连接不是数据, 如何序列化?
     * userId - Channel
     */
    private ConcurrentMap<Long, Channel> USER_CHANNEL = new ConcurrentHashMap<>();

    private ConcurrentMap<ChannelId, Long> CHANNEL_USER = new ConcurrentHashMap<>();

    /**
     * # 添加 Channel 到 {@link #channels} 中
     *
     * @param channel Channel
     */
    public void add(Channel channel) {
        channels.put(channel.id(), channel);
        log.debug("连接 [{}] 已上线", channel.id());
    }

    /**
     *
     * @param channel   /
     * @param userId
     */
    public void addUser(Channel channel, Long userId) {
        Channel existChannel = channels.get(channel.id());
        if (existChannel == null) {
            log.debug("连接 [{}] 不存在", channel.id());
            return;
        }
        // 设置属性
        channel.attr(CHANNEL_ATTR_KEY_USER).set(userId);
        USER_CHANNEL.put(userId, channel);
        CHANNEL_USER.put(channel.id(), userId);
    }

    /**
     *
     * @param channel
     */
    public void remove(Channel channel) {
        log.debug("连接 [{}] 已断开连接", channel.id());
        // 移除 channels
        channels.remove(channel.id());

        if (USER_CHANNEL.size() < 1) {
            return;
        }

        USER_CHANNEL.remove(CHANNEL_USER.get(channel.id()));
        CHANNEL_USER.remove(channel.id());
    }

    /**
     * 向指定用户发送消息
     *
     * @param userId       用户
     * @param invocation 消息体
     */
    public void send(Long userId, Invocation invocation) {
        // 获得用户对应的 Channel
        Channel channel = USER_CHANNEL.get(userId);
        // TODO 用户不在线, 该如何实现?
        if (channel == null) {
            log.debug("连接不存在");
            return;
        }
        if (!channel.isActive()) {
            log.debug("连接 [{}] 未被激活", channel.id());
            return;
        }
        // 发送消息
        TextWebSocketFrame frame = new TextWebSocketFrame(JSON.toJSONString(invocation));
        channel.writeAndFlush(frame);
    }

    /**
     * 群聊 [排除当前发送人]
     *
     * @param groupId
     * @param chatRedirect
     */
    public void sendGroup(Long groupId, ChatRedirect chatRedirect) {
        Invocation invocation = new Invocation(ChatRedirect.TYPE, chatRedirect);
        // 获取群聊中的所有人, 不包括发送人
        R result = groupMemberFeignService.findGroupMemberByGroupId(groupId, Long.valueOf(chatRedirect.getSendId()));
        if ((((List) result.getData())).size() < 1) {
            return;
        }

        ((List) result.getData()).stream().forEach(item -> {
            JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(item));
            Channel channel = USER_CHANNEL.get(jsonObject.getLong("userId"));
            if (!channel.isActive()) {
                log.debug("连接 [{}] 未被激活", channel.id());
                return;
            }
            chatRedirect.setReceiveId(jsonObject.getString("userId"));
            invocation.setMessage(JSON.toJSONString(chatRedirect));
            TextWebSocketFrame frame = new TextWebSocketFrame(JSON.toJSONString(invocation));
            channel.writeAndFlush(frame);
        });
    }

}
